package com.surfilter.msmp.web.es;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.flush.FlushRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Description：ES索引管理
 * Author；JinHuatao
 * Date: 2019/10/8 11:14
 */
@Component
public class EsIndexManager implements IIndex{

    private static final Logger logger = LoggerFactory.getLogger(EsIndexManager.class);

    @Autowired
    private RestHighLevelClient highLevelClient;

    @Override
    public void insertData(String indexName, Map<String, Object> sourceMap) {
        IndexRequest indexRequest = new IndexRequest(indexName).source(sourceMap);
        try {
            highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("插入ES数据异常，{}", e);
        }
    }

    @Override
    public void insertJsonData(String indexName, String sourceJson) {
        IndexRequest indexRequest = new IndexRequest(indexName).source(sourceJson);
        try {
            highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("插入ES数据异常，{}", e);
        }
    }

    private boolean updateDocument(IndexRequest indexRequest){
        try {
            IndexResponse indexResponse = highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED
                    || indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                return true;
            }

            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                return true;
            }

            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    logger.error(failure.reason());
                }
            }
        } catch (IOException e) {
            logger.error("插入ES数据异常，{}", e);
        }
        return false;
    }

    @Override
    public boolean insertData(String indexName, Map<String, Object> sourceMap, String id) {
        IndexRequest indexRequest = new IndexRequest(indexName, "_doc", id).source(sourceMap);
        return updateDocument(indexRequest);
    }

    @Override
    public boolean insertJsonData(String indexName, String sourceJson, String id) {
        IndexRequest indexRequest = new IndexRequest(indexName,"_doc", id).source(sourceJson, XContentType.JSON);
        return updateDocument(indexRequest);
    }

    @Override
    public void batchInsertData(String index, Map<String, Map<String, Object>> sourceMap) {
        try {
            BulkRequest request = new BulkRequest();
            for(String id: sourceMap.keySet()){
                request.add(new IndexRequest(index, "_doc").opType("create").id(id).source(sourceMap.get(id)));
            }
            highLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("批量插入ES数据异常，{}", e);
        }
    }

    @Override
    public void batchInsertData(String index, List<Map<String, Object>> sourceList) {
        try {
            BulkRequest request = new BulkRequest();
            for(Map<String, Object> source: sourceList){
                request.add(new IndexRequest(index,  "_doc").opType("create").source(source));
            }
            highLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("批量插入ES数据异常，{}", e);
        }
    }

    @Override
    public void deleteByQuery(String indexName, QueryBuilder query) {
        DeleteByQueryRequest request = new DeleteByQueryRequest(indexName);
        request.setQuery(query);
        try {
            highLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("删除ES索引{}数据异常，{}", indexName, e);
        }
    }

    @Override
    public void deleteIndexDoc(String indexName, String id) {
        DeleteRequest request = new DeleteRequest(indexName, "_doc", id);
        try {
            highLevelClient.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("删除ES索引{}数据异常，{}", indexName, e);
        }
    }

    @Override
    public void deleteIndexDoc(String indexName, List<String> ids) {
        BulkRequest bulkRequest = new BulkRequest();
        for(String id: ids){
            bulkRequest.add(new DeleteRequest(indexName, "_doc", id));
        }
        try {
            highLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("批量删除ES索引{}数据异常，{}", indexName, e);
        }
    }

    @Override
    public void optimize(String indexName) {
        //TODO 未找到API
    }

    @Override
    public boolean indexExists(String indexName) {
        try {
            GetIndexRequest request = new GetIndexRequest(indexName);
            request.local(false);
            request.humanReadable(true);
            return highLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("获取ES索引信息异常，{}", e);
        }
        return false;
    }

    private boolean updateIndex(String indexName, CreateIndexRequest request){
        try {
            CreateIndexResponse createIndexResponse = highLevelClient.indices().create(request, RequestOptions.DEFAULT);
            boolean acknowledged = createIndexResponse.isAcknowledged();
            logger.info("索引" + indexName + "创建成功！");
            if(acknowledged)
                return true;
        } catch (IOException e) {
            logger.error("索引" + indexName + "创建失败,{}", e);
        }
        return false;
    }

    public boolean createIndex(String indexName){
        return createIndex(indexName, 5, 1);
    }

    @Override
    public boolean createIndex(String indexName, int shards, int replicas) {
        if(indexExists(indexName)){
            return true;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder().put("index.number_of_shards", shards)
                .put("index.number_of_replicas", replicas));
       return updateIndex(indexName, request);
    }

    @Override
    public boolean createIndex(String indexName, Map<String, Object> properties) {
        if(indexExists(indexName)){
            return true;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder().put("index.number_of_shards", 5)
                .put("index.number_of_replicas", 1));
        Map<String, Object> mapping = new HashMap<>();
        mapping.put("properties", properties);
        request.mapping(mapping);
        return updateIndex(indexName, request);
    }

    @Override
    public boolean createIndex(String indexName, final XContentBuilder mapping) {
        if(indexExists(indexName)){
            return true;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder().put("index.number_of_shards", 5)
                .put("index.number_of_replicas", 1));
        request.mapping(mapping);
        return updateIndex(indexName, request);
    }

    @Override
    public boolean updateIndexRefreshInterval(String indexName, int interval) {
        UpdateSettingsRequest request = new UpdateSettingsRequest(indexName);
        request.settings(Settings.builder().put("index.refresh_interval", interval));
        try {
            return highLevelClient.indices().putSettings(request, RequestOptions.DEFAULT).isAcknowledged();
        } catch (IOException e) {
            logger.error("索引" + indexName + "刷新配置更新失败,{}", e);
        }
        return false;
    }

    @Override
    public boolean updateIndexRepliceas(String indexName, int repliceas) {
        UpdateSettingsRequest request = new UpdateSettingsRequest(indexName);
        request.settings(Settings.builder().put("index.number_of_replicas", repliceas)
                .put("index.translog.flush_threshold_ops", "10000")
                .put("index.refresh_interval", "1"));
        try {
            return highLevelClient.indices().putSettings(request, RequestOptions.DEFAULT).isAcknowledged();
        } catch (IOException e) {
            logger.error("索引" + indexName + "刷新配置更新失败,{}", e);
        }
        return false;
    }

    @Override
    public boolean createIndexAliasName(String indexName, String aliasName) {
        IndicesAliasesRequest.AliasActions aa = IndicesAliasesRequest.AliasActions.add();
        aa.index(indexName).alias(aliasName);
        IndicesAliasesRequest request = new IndicesAliasesRequest();
        request.addAliasAction(aa);
        try {
            AcknowledgedResponse response = highLevelClient.indices().updateAliases(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            logger.error("索引" + indexName + "别名创建失败,{}", e);
        }
        return false;
    }

    @Override
    public boolean removeAlias(String indexName, String aliasName) {
        IndicesAliasesRequest.AliasActions aa = IndicesAliasesRequest.AliasActions.remove();
        aa.index(indexName).alias(aliasName);
        IndicesAliasesRequest request = new IndicesAliasesRequest();
        request.addAliasAction(aa);
        try {
            AcknowledgedResponse response = highLevelClient.indices().updateAliases(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            logger.error("删除索引" + indexName + "别名失败,{}", e);
        }
        return false;
    }

    @Override
    public boolean changeAlias(String srcIndexName, String destIndexName, String indexAlias) {
        /*IndicesAliasesRequest.AliasActions aa = IndicesAliasesRequest.AliasActions.add();
        aa.index(srcIndexName).alias(indexAlias);
        IndicesAliasesRequest request = new IndicesAliasesRequest();
        request.addAliasAction(aa);
        try {
            AcknowledgedResponse response = highLevelClient.indices().updateAliases(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            e.printStackTrace();
        }*/
        //TODO 修改索引别名
        return false;
    }

    @Override
    public boolean closeIndex(String indexName) {
        CloseIndexRequest request = new CloseIndexRequest(indexName);
        try {
            AcknowledgedResponse response = highLevelClient.indices().close(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            logger.error("关闭ES索引异常，{}", e);
        }
        return false;
    }

    @Override
    public boolean openIndex(String indexName) {
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        try {
            AcknowledgedResponse response = highLevelClient.indices().open(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            logger.error("打开ES索引异常，{}", e);
        }
        return false;
    }

    @Override
    public boolean deleteIndex(String indexName) {
        try {
            DeleteIndexRequest request = new DeleteIndexRequest(indexName);
            AcknowledgedResponse deleteIndexResponse = highLevelClient.indices().delete(request, RequestOptions.DEFAULT);
            return deleteIndexResponse.isAcknowledged();
        } catch (ElasticsearchException exception) {
            if (exception.status() == RestStatus.NOT_FOUND) {
                return true;
            } else {
                return false;
            }
        } catch (IOException e) {
            logger.error("删除ES索引信息异常，{}", e);
        }
        return false;
    }

    @Override
    public void flushIndex(String indexName) {
        FlushRequest request = new FlushRequest(indexName);
        try {
            highLevelClient.indices().flush(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("刷新ES索引失败，{}", e);
        }
    }

    @Override
    public SearchResponse prepareSearch(String indexName, long scrollTime, int limit) {
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.searchType(SearchType.DEFAULT)
                .scroll(TimeValue.timeValueHours(scrollTime))
                .setBatchedReduceSize(limit);
        try {
           return highLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("预处理查询ES索引失败，{}", e);
        }
        return null;
    }

    @Override
    public SearchResponse searchScroll(String scrollId, long scrollTime) {
        SearchScrollRequest ssr = new SearchScrollRequest(scrollId);
        ssr.scroll(TimeValue.timeValueHours(scrollTime));
        try {
            return highLevelClient.scroll(ssr, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("滚动查询ES索引失败，{}", e);
        }
        return null;
    }
}
